package kafka

import (
	"time"

	"gitee.com/zawei/yggdrasil/pkg/config"
	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/Shopify/sarama"
)

type GroupConsumerConfig struct {
	Brokers            []string
	Topics             []string
	GroupID            string
	AutoCommitInterval time.Duration
	Version            string
	BalanceStrategy    string
	EnableTrace        bool
	OffsetStrategy     string

	handler      MessageHandler
	interceptors []sarama.ConsumerInterceptor
}

func StdGroupConsumerConfig() *GroupConsumerConfig {
	return RawGroupConsumerConfig("kafka.group_consumer")
}

func RawGroupConsumerConfig(key string) *GroupConsumerConfig {
	c := new(GroupConsumerConfig)
	if err := config.Get(key).Scan(c); err != nil {
		log.Fatalf("fault to scan config[%s], error: %s", key, err.Error())
	}
	return c
}

func (cfg *GroupConsumerConfig) WithHandler(handler MessageHandler) {
	cfg.handler = handler
}

func (cfg *GroupConsumerConfig) WithInterceptors(interceptors ...sarama.ConsumerInterceptor) {
	if cfg.interceptors == nil {
		cfg.interceptors = make([]sarama.ConsumerInterceptor, 0, len(interceptors))
	}
	cfg.interceptors = append(cfg.interceptors, interceptors...)
}

func (cfg *GroupConsumerConfig) Build() (consumer *Consumer, err error) {
	if cfg.handler == nil {
		cfg.handler = messageHandler{}
	}

	c := sarama.NewConfig()
	if cfg.Version != "" {
		c.Version, err = sarama.ParseKafkaVersion(cfg.Version)
		if err != nil {
			return
		}
	} else {
		c.Version = sarama.MaxVersion
	}
	switch cfg.BalanceStrategy {
	case "range":
		c.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
	case "roundRobin":
		c.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	case "sticky":
		c.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
	}

	consumer = &Consumer{
		GroupConsumerConfig: cfg,
	}
	switch cfg.OffsetStrategy {
	case "sequence":
		consumer.offsetManager = newSequenceManager(cfg.lg)
	default:
		consumer.offsetManager = &defaultManager{}
	}
	consumer.group, err = sarama.NewConsumerGroup(cfg.Brokers, cfg.GroupID, c)
	if err != nil {
		return nil, err
	}

	return
}

type ProducerConfig struct {
	Brokers      []string
	Sync         bool
	Version      string
	RequiredAcks string
	EnableTrace  bool

	interceptors []sarama.ProducerInterceptor
	errHandle    func(*sarama.ProducerError)
	sucHandle    func(*sarama.ProducerMessage)
}

func StdProducerConfig() *ProducerConfig {
	return RawProducerConfig("kafka.producer")
}

func RawProducerConfig(key string) *ProducerConfig {
	c := new(ProducerConfig)
	if err := config.Get(key).Scan(c); err != nil {
		log.Fatalf("fault to scan config[%s], error: %s", key, err.Error())
	}
	return c
}

func (cfg *ProducerConfig) WithErrHandle(errHandle func(*sarama.ProducerError)) {
	cfg.errHandle = errHandle
}

func (cfg *ProducerConfig) WithSucHandle(sucHandle func(*sarama.ProducerMessage)) {
	cfg.sucHandle = sucHandle
}

func (cfg *ProducerConfig) WithInterceptors(interceptors ...sarama.ProducerInterceptor) {
	if cfg.interceptors == nil {
		cfg.interceptors = make([]sarama.ProducerInterceptor, 0, len(interceptors))
	}
	cfg.interceptors = append(cfg.interceptors, interceptors...)
}

func (cfg *ProducerConfig) Build() (producer *Producer, err error) {
	kafkaCfg := sarama.NewConfig()
	if cfg.Version != "" {
		kafkaCfg.Version, err = sarama.ParseKafkaVersion(cfg.Version)
		if err != nil {
			return
		}
	} else {
		kafkaCfg.Version = sarama.MaxVersion
	}
	switch cfg.RequiredAcks {
	case "all":
		kafkaCfg.Producer.RequiredAcks = sarama.WaitForAll
	case "local":
		kafkaCfg.Producer.RequiredAcks = sarama.WaitForLocal
	case "no":
		kafkaCfg.Producer.RequiredAcks = sarama.NoResponse
	}
	kafkaCfg.Producer.Return.Successes = true
	kafkaCfg.Producer.Return.Errors = true

	kafkaCfg.Producer.Interceptors = append(kafkaCfg.Producer.Interceptors, cfg.interceptors...)
	producer = &Producer{
		ProducerConfig: cfg,
	}
	if cfg.Sync {
		producer.SyncProducer, err = sarama.NewSyncProducer(cfg.Brokers, kafkaCfg)
	} else {
		producer.AsyncProducer, err = sarama.NewAsyncProducer(cfg.Brokers, kafkaCfg)
	}
	if err != nil {
		return
	}

	return
}
